package com.atguigu.gmall.realtime.util;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * @author caodan
 * @version 1.0
 * @Deacription
 * @date 2022-09-09 16:04
 */
public class KafkaUtil {

    public static final String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092,Hadoop104:9092";

    /**
     * 获取kafka工具类对象
     *
     * @param topicName 主题
     * @param groupId   消费者组
     * @return 对象
     */
    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topicName, String groupId) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        return new FlinkKafkaConsumer<>(topicName, new KafkaDeserializationSchema<String>() {
            @Override
            public boolean isEndOfStream(String s) {
                return false;
            }

            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> consumerRecord) throws Exception {
                if (consumerRecord != null && consumerRecord.value() != null) {
                    return new String(consumerRecord.value());
                }
                return null;
            }

            @Override
            public TypeInformation<String> getProducedType() {
                return TypeInformation.of(String.class);
            }
        }, props);
    }

    /**
     * 获取kafka发送端
     *
     * @param topicName 主题名称
     * @return 消费者
     */
    public static FlinkKafkaProducer<String> getFlinkKafkaProducer(String topicName) {

        Properties porps = new Properties();
        porps.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVER);
        porps.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, String.valueOf(15 * 60 * 1000));

        return new FlinkKafkaProducer<>("default-topic", (KafkaSerializationSchema<String>) (value, aLong)
                -> new ProducerRecord<>(topicName, value.getBytes(StandardCharsets.UTF_8)), porps, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }


    /**
     * 从kafka中获取连接器属性
     *
     * @param topicName 主题名
     * @param groupId   组名
     * @return 属性
     */
    public static String getKafkaDDLPorps(String topicName, String groupId) {
        return "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topicName + "',\n" +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
                "  'properties.group.id' = '" + groupId + "',\n" +
                "  'scan.startup.mode' = 'group-offsets',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    /**
     * 获取upset-kafka连接参数属性
     *
     * @param topicName 主题名
     * @return 字符串
     */
    public static String getUpsetKafkaProps(String topicName) {
        return "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '" + topicName + "',\n" +
                "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
    }

    /**
     *  从topic_db 中获取数据
     * @param groupId 消费者组
     * @return 结果
     */
    public static String getTopicDbDDL(String groupId) {
        return "create table topic_db(" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<String, String>,\n" +
                "`old` map<String, String>,\n" +
                "`proc_time` as PROCTIME(),\n" +
                "`ts` string\n" +
                ") with ('connector' = 'kafka', " +
                " 'topic' = 'topic_db'," +
                " 'properties.bootstrap.servers' = '" + KAFKA_SERVER + "', " +
                " 'properties.group.id' = '" + groupId + "', " +
                " 'format' = 'json', " +
                " 'scan.startup.mode' = 'group-offsets')";
    }
}
